//! The main entry point of this crate, composed of three parts:
//!
//! - `Client`: the client side of a p2p node, which is used to send requests to a remote
//!   peer or broadcast messages to the network.
//!
//! - `Server`: the server side of a p2p node, which is used to handle requests / broadcast-messages
//!   from remote peers.
//!
//! - `EventHandler`: the trait that defines how to handle requests / broadcast-messages from remote peers.
//!   The application should implement this trait and pass it to the `Server`.

use std::{cell::OnceCell, collections::HashMap, fmt::Debug, io, time::Duration};

use itertools::Itertools;
use libp2p::{
    autonat,
    bytes::BufMut,
    futures::prelude::*,
    gossipsub::{self, TopicHash},
    identify,
    identity::ed25519,
    identity::{self, Keypair},
    kad::{self, store::MemoryStore, QueryResult},
    mdns, noise, ping,
    request_response::{self, OutboundFailure, OutboundRequestId, ResponseChannel},
    swarm::{DialError, SwarmEvent},
    tcp, yamux, Multiaddr, PeerId, StreamProtocol, Swarm, SwarmBuilder,
};

use log::{error, info, warn};
use tokio::{
    select,
    sync::{
        mpsc::{self, UnboundedReceiver, UnboundedSender},
        oneshot,
    },
    time::{self, Interval},
};

use crate::{
    config::P2pConfig,
    error::P2pError,
    protocol::{self, *},
    transport, PeerIdWithMultiaddr,
};

const BOOTNODES: [&str; 5] = [
    "/dnsaddr/bootstrap.libp2p.io/p2p/QmNnooDu7bfjPFoTZYxMNLWUQJyrVwtbZg5gBMjTezGAJN",
    "/dnsaddr/bootstrap.libp2p.io/p2p/QmQCU2EcMqAqQPR2i9bChDtGNJchTbq5TbXJJ16u19uLTa",
    "/dnsaddr/bootstrap.libp2p.io/p2p/QmbLHAnMoJPWSCR5Zhtx6BHJX9KiKNN6tpvbUcqanj75Nb",
    "/dnsaddr/bootstrap.libp2p.io/p2p/QmcZf59bWwK5XFi76CZX8cbJ4BhTzzA3gU1ZjYZcYW3dwt",
    "/ip4/104.131.131.82/tcp/4001/p2p/QmaCpDMGvV2BGHeYERUEnRQAwe3N8SzbUtfsmvsqQLuvuJ",
];

/// `EventHandler` is the trait that defines how to handle requests / broadcast-messages from remote peers.
pub trait EventHandler: Debug + Send + 'static {
    /// Handles an inbound request from a remote peer.
    fn handle_inbound_request(&self, request: Vec<u8>) -> Result<Vec<u8>, P2pError>;

    /// Handles an broadcast message from a remote peer.
    fn handle_broadcast(&self, topic: &str, message: Vec<u8>);
}

#[derive(Clone, Debug)]
pub struct Client {
    cmd_sender: UnboundedSender<Command>,
}

pub struct Server<E: EventHandler> {
    /// The actual network service.
    network_service: Swarm<QuantsBehaviour>,

    /// The local peer id.
    local_peer_id: PeerId,
    /// The addresses that the server is listening on.
    listened_addresses: Vec<Multiaddr>,
    /// The receiver of commands from the client.
    cmd_receiver: UnboundedReceiver<Command>,
    /// The handler of events from remote peers.
    event_handler: OnceCell<E>,

    /// The ticker to periodically discover new peers.
    discovery_ticker: Interval,
    /// The pending outbound requests, awaiting for a response from the remote.
    pending_outbound_requests: HashMap<OutboundRequestId, oneshot::Sender<ResponseType>>,
    /// The topics will be hashed when subscribing to the gossipsub protocol,
    /// but we need to keep the original topic names for broadcasting.
    pubsub_topics: Vec<String>,
}

/// Create a new secret key for the p2p node.
pub fn new_secret_key() -> String {
    let secret = ed25519::SecretKey::generate();
    bs58::encode(secret.as_ref()).into_string()
}

/// Create a new p2p node, which consists of a `Client` and a `Server`.
pub fn new<E: EventHandler>(config: P2pConfig) -> Result<(Client, Server<E>), P2pError> {
    let (cmd_sender, cmd_receiver) = mpsc::unbounded_channel();

    let server = Server::new(config, cmd_receiver)?;
    let client = Client { cmd_sender };

    Ok((client, server))
}

impl Client {
    /// Send a blocking request to the `target` peer.
    pub fn blocking_request(&self, target: &str, request: Vec<u8>) -> Result<Vec<u8>, P2pError> {
        let target = target.parse().map_err(|_| P2pError::InvalidPeerId)?;

        let (responder, receiver) = oneshot::channel();
        let _ = self.cmd_sender.send(Command::SendRequest {
            target,
            request,
            responder,
        });
        receiver
            .blocking_recv()?
            .map_err(|_| P2pError::RequestRejected)
    }

    /// Publish a message to the given topic.
    pub fn broadcast(&self, topic: impl Into<String>, message: Vec<u8>) {
        let _ = self.cmd_sender.send(Command::Broadcast {
            topic: topic.into(),
            message,
        });
    }

    /// Get known peers of the node.
    pub fn get_known_peers(&self) -> Vec<String> {
        self.get_node_status()
            .known_peers
            .into_keys()
            .map(|id| id.to_base58())
            .collect()
    }

    /// Get status of the node for debugging.
    pub fn get_node_status(&self) -> NodeStatus {
        let (responder, receiver) = oneshot::channel();
        let _ = self.cmd_sender.send(Command::GetStatus(responder));
        receiver.blocking_recv().unwrap_or_default()
    }
}

/// The commands sent by the `Client` to the `Server`.
pub enum Command {
    SendRequest {
        target: PeerId,
        request: Vec<u8>,
        responder: oneshot::Sender<ResponseType>,
    },
    Broadcast {
        topic: String,
        message: Vec<u8>,
    },
    GetStatus(oneshot::Sender<NodeStatus>),
}

impl<E: EventHandler> Server<E> {
    /// Create a new `Server`.
    pub fn new(
        config: P2pConfig,
        cmd_receiver: UnboundedReceiver<Command>,
    ) -> Result<Self, P2pError> {
        let addr: Multiaddr = config.addr.parse()?;
        let local_key = config.gen_keypair()?;
        let local_peer_id = local_key.public().to_peer_id();
        info!("\n\n📣 Local peer id: {local_peer_id:?} \n\n");

        let pubsub_topics = config.pubsub_topics;

        let behaviour = QuantsBehaviour::new(
            local_key.clone(),
            pubsub_topics.clone(),
            config.req_resp.clone(),
        )?;

        let mut swarm = libp2p::SwarmBuilder::with_existing_identity(local_key.clone())
            .with_tokio()
            .with_quic()
            .with_other_transport(|_| transport::build_transport(local_key.clone()))?
            // .with_tcp(
            //     tcp::Config::default().port_reuse(true).nodelay(true),
            //     noise::Config::new,
            //     yamux::Config::default,
            // )?
            .with_dns()?
            // .with_relay_client(noise::Config::new, yamux::Config::default)?
            .with_behaviour(|_keypair| behaviour)?
            .build();

        // Switch to server mode.
        swarm.add_external_address(addr.clone());

        swarm.listen_on(addr)?;

        // Connect to the boot node if specified.
        if let Some(boot_node) = config.boot_node {
            swarm.dial(boot_node.address())?;
        }

        // Add the bootnodes to the local routing table. `libp2p-dns` built
        // into the `transport` resolves the `dnsaddr` when Kademlia tries
        // to dial these nodes.
        for addr in &BOOTNODES {
            let boot_node: Option<PeerIdWithMultiaddr> = addr.parse().ok();
            if let Some(bn) = boot_node {
                swarm.dial(bn.address())?
            }
        }

        // Create a ticker to periodically discover new peers.
        let interval_secs = config.discovery_interval.unwrap_or(30);
        let instant = time::Instant::now() + Duration::from_secs(5);
        let discovery_ticker = time::interval_at(instant, Duration::from_secs(interval_secs));

        Ok(Self {
            network_service: swarm,
            local_peer_id,
            listened_addresses: Vec::new(),
            cmd_receiver,
            event_handler: OnceCell::new(),
            discovery_ticker,
            pending_outbound_requests: HashMap::new(),
            pubsub_topics,
        })
    }

    /// Set the handler of events from remote peers.
    pub fn set_event_handler(&mut self, handler: E) {
        self.event_handler.set(handler).unwrap();
    }

    /// Run the `Server`.
    pub async fn run(mut self) {
        loop {
            select! {
                // Next discovery process.
                _ = self.discovery_ticker.tick() => {
                    self.network_service.behaviour_mut().discover_peers();
                },

                // Next command from the `Client`.
                msg = self.cmd_receiver.recv() => {
                    if let Some(cmd) = msg {
                        self.handle_command(cmd);
                    }
                },
                // Next event from `Swarm`.
                event = self.network_service.select_next_some() => {
                    self.handle_swarm_event(event);
                },
            }
        }
    }

    // Process the next command coming from `Client`.
    fn handle_command(&mut self, cmd: Command) {
        match cmd {
            Command::SendRequest {
                target,
                request,
                responder,
            } => self.handle_outbound_request(target, request, responder),
            Command::Broadcast { topic, message } => self.handle_outbound_broadcast(topic, message),
            Command::GetStatus(responder) => responder.send(self.get_status()).unwrap(),
        }
    }

    // Process the next event coming from `Swarm`.
    fn handle_swarm_event(&mut self, event: SwarmEvent<QuantsBehaviourEvent>) {
        let behaviour_ev = match event {
            SwarmEvent::Behaviour(ev) => ev,

            SwarmEvent::NewListenAddr { address, .. } => {
                info!("📣 P2P node listening on {:?}", address);
                return self.update_listened_addresses();
            }

            SwarmEvent::ListenerClosed {
                reason, addresses, ..
            } => return Self::log_listener_close(reason, addresses),

            // Can't connect to the `peer`, remove it from the DHT.
            SwarmEvent::OutgoingConnectionError {
                peer_id: Some(peer_id),
                error,
                ..
            } => {
                log::debug!(
                    "Can't connect to the `peer` {}, remove it from the DHT. ",
                    peer_id
                );
                return self.network_service.behaviour_mut().remove_peer(&peer_id);
            }

            _ => return,
        };

        self.handle_behaviour_event(behaviour_ev);
    }

    fn handle_behaviour_event(&mut self, ev: QuantsBehaviourEvent) {
        match ev {
            // See https://docs.rs/libp2p/latest/libp2p/kad/index.html#important-discrepancies
            QuantsBehaviourEvent::Identify(identify::Event::Received { peer_id, info }) => {
                log::debug!("*****identify found peer_id= {peer_id}");
                if protocol::wellknown_peers().contains(&peer_id.to_string().as_str()) {
                    log::info!("\n\n**NNNN** Identify peer_id= {peer_id}  {info:?}");

                    let pc = StreamProtocol::new("/libp2p/circuit/relay/0.2.0/hop");
                    if info.protocols.contains(&pc) {
                        log::debug!("\n\n\n~~~~ peer_id= {peer_id} {info:?}")
                    }
                }

                self.add_addresses(&peer_id, info.listen_addrs)
            }

            // The remote peer is unreachable, remove it from the DHT.
            QuantsBehaviourEvent::Ping(ping::Event {
                peer,
                result: Err(_),
                ..
            }) => self.network_service.behaviour_mut().remove_peer(&peer),

            QuantsBehaviourEvent::ReqResp(request_response::Event::Message {
                peer,
                message:
                    request_response::Message::Request {
                        request, channel, ..
                    },
                ..
            }) => self.handle_inbound_request(peer, request, channel),

            QuantsBehaviourEvent::ReqResp(request_response::Event::Message {
                peer,
                message:
                    request_response::Message::Response {
                        request_id,
                        response,
                    },
                ..
            }) => self.handle_inbound_response(peer, request_id, response),

            QuantsBehaviourEvent::ReqResp(request_response::Event::OutboundFailure {
                request_id,
                error,
                ..
            }) => self.handle_outbound_failure(request_id, error),

            QuantsBehaviourEvent::Pubsub(gossipsub::Event::Message {
                propagation_source: _,
                message_id: _,
                message,
            }) => self.handle_inbound_broadcast(message),
            QuantsBehaviourEvent::Kad(kad::Event::OutboundQueryProgressed {
                id,
                result,
                stats,
                step,
            }) => {
                match result {
                    QueryResult::GetClosestPeers(ref res) => {
                        if let Ok(ok_result) = res {
                            let key_str = bs58::encode(&ok_result.key)
                                .with_alphabet(bs58::Alphabet::RIPPLE)
                                .into_string();
                            let peers = &ok_result.peers;
                            // Now you can use the key here
                            log::debug!(
                                "\nn\n****9999 GetClosestPeers  key={key_str} {peers:?}\n\n"
                            );
                        } else {
                            // Handle the error case if needed
                            log::error!("Error occurred in GetClosestPeersResult: {res:?}");
                        }
                    }
                    _ => {
                        // Handle other variants of QueryResult if needed
                        log::info!("Other QueryResult variant: {result:?}");
                    }
                }

                log::debug!(
                    "\n\nKad todo!() OutboundQueryProgressed id={id}, result={result:?} stats={stats:?} step={step:?} \n\n"
                );
                // log::error!("Kad todo!() {result:?}");
            }
            QuantsBehaviourEvent::AutoNat(evt) => match evt {
                autonat::Event::StatusChanged { old, new } => {
                    log::error!("\n AutoNat ====  StatusChanged! old= {old:?} , new= {new:?}\n")
                }
                autonat::Event::InboundProbe(ipe) => {
                    log::debug!("\n AutoNat <<<<  InboundProbe!  {ipe:?}\n")
                }
                autonat::Event::OutboundProbe(ope) => {
                    log::debug!("\n AutoNat >>>>  OutboundProbe!  {ope:?}\n")
                }
            },
            /*
                        BehaviourEvent::Mdns(mdns::Event::Discovered(list)) => {
                            for (peer_id, multiaddr) in list {
                                info!("mDNS discovered a new peer: {peer_id}, {multiaddr}");
                                // swarm.behaviour_mut().gossipsub.add_explicit_peer(&peer_id);
                                self.network_service
                                    .behaviour_mut()

                                    .add_address(&peer_id, multiaddr);
                            }
                        }
                        BehaviourEvent::Mdns(mdns::Event::Expired(list)) => {
                            for (peer_id, _multiaddr) in list {
                                info!("mDNS discover peer has expired: {peer_id}");
                                // swarm
                                //     .behaviour_mut()
                                //     .gossipsub
                                //     .remove_explicit_peer(&peer_id);
                                self.network_service.behaviour_mut().remove_peer(&peer_id);
                            }
                        }
            */
            // BehaviourEvent::AutoNat(autonat::Event::InboundProbe(evt)) => {
            //     info!("autoNAT event={evt:?}");
            // }

            // BehaviourEvent::AutoNat(autonat::Event::OutboundProbe(evt)) => {
            //     info!("autoNAT event={evt:?}");
            // }

            // BehaviourEvent::AutoNat(autonat::Event::StatusChanged { old, new }) => {
            //     info!("autoNAT old={old:?} , new={new:?}");
            // }
            _ => {}
        }
    }

    // Inbound requests are handled by the `EventHandler` which is provided by the application layer.
    fn handle_inbound_request(
        &mut self,
        peer_id: PeerId,
        request: Vec<u8>,
        ch: ResponseChannel<ResponseType>,
    ) {
        if let Some(handler) = self.event_handler.get() {
            let response = handler.handle_inbound_request(request).map_err(|_| ());
            self.network_service
                .behaviour_mut()
                .send_response(ch, response);
        }
    }

    // Store the request_id with the responder so that we can send the response later.
    fn handle_outbound_request(
        &mut self,
        target: PeerId,
        request: Vec<u8>,
        responder: oneshot::Sender<ResponseType>,
    ) {
        let req_id = self
            .network_service
            .behaviour_mut()
            .send_request(&target, request);
        self.pending_outbound_requests.insert(req_id, responder);
    }

    // An outbound request failed, notify the application layer.
    fn handle_outbound_failure(&mut self, request_id: OutboundRequestId, error: OutboundFailure) {
        if let Some(responder) = self.pending_outbound_requests.remove(&request_id) {
            info!("❌ Outbound request failed: {:?}", error);
            let _ = responder.send(Err(()));
        } else {
            warn!("❗ Received failure for unknown request: {}", request_id);
            debug_assert!(false);
        }
    }

    // An inbound response was received, notify the application layer.
    fn handle_inbound_response(
        &mut self,
        peer_id: PeerId,
        request_id: OutboundRequestId,
        response: ResponseType,
    ) {
        if let Some(responder) = self.pending_outbound_requests.remove(&request_id) {
            let _ = responder.send(response);
        } else {
            warn!("❗ Received response for unknown request: {}", request_id);
            debug_assert!(false);
        }
    }

    // Inbound broadcasts are handled by the `EventHandler` which is provided by the application layer.
    fn handle_inbound_broadcast(&mut self, message: gossipsub::Message) {
        if let Some(handler) = self.event_handler.get() {
            let topic_hash = message.topic;
            match self.get_topic(&topic_hash) {
                Some(topic) => handler.handle_broadcast(&topic, message.data),
                None => {
                    warn!("❗ Received broadcast for unknown topic: {:?}", topic_hash);
                    debug_assert!(false);
                }
            }
        }
    }

    // Broadcast a message to all peers subscribed to the given topic.
    fn handle_outbound_broadcast(&mut self, topic: String, message: Vec<u8>) {
        let _ = self
            .network_service
            .behaviour_mut()
            .broadcast(topic, message);
    }

    fn add_addresses(&mut self, peer_id: &PeerId, addresses: Vec<Multiaddr>) {
        for addr in addresses.into_iter().unique() {
            self.network_service
                .behaviour_mut()
                .add_address(peer_id, addr);
        }
    }

    fn get_status(&mut self) -> NodeStatus {
        let known_peers = self.network_service.behaviour_mut().known_peers();
        NodeStatus {
            local_peer_id: self.local_peer_id.to_base58(),
            listened_addresses: self.listened_addresses.clone(),
            known_peers_count: known_peers.len(),
            known_peers,
        }
    }

    fn update_listened_addresses(&mut self) {
        self.listened_addresses = self
            .network_service
            .listeners()
            .map(ToOwned::to_owned)
            .collect();
    }

    /// Returns the topic name for the given topic hash.
    fn get_topic(&self, topic_hash: &TopicHash) -> Option<String> {
        for t in &self.pubsub_topics {
            let topic = gossipsub::IdentTopic::new(t);
            if topic.hash() == *topic_hash {
                return Some(t.clone());
            }
        }

        None
    }

    fn log_listener_close(reason: io::Result<()>, addresses: Vec<Multiaddr>) {
        let addrs = addresses
            .into_iter()
            .map(|a| a.to_string())
            .collect::<Vec<_>>()
            .join(", ");
        match reason {
            Ok(()) => {
                info!("📣 Listener ({}) closed gracefully", addrs)
            }
            Err(e) => {
                error!("❌ Listener ({}) closed: {}", addrs, e)
            }
        }
    }
}

/// The node status, for debugging.
#[derive(Clone, Debug, Default)]
pub struct NodeStatus {
    pub local_peer_id: String,
    pub listened_addresses: Vec<Multiaddr>,
    pub known_peers_count: usize,
    pub known_peers: HashMap<PeerId, Vec<Multiaddr>>,
}

#[cfg(test)]
mod tests {
    use super::*;

    #[test]
    fn new_secret_key_works() {
        let key1 = new_secret_key();
        let key2 = new_secret_key();
        assert_ne!(key1, key2);

        let secret = bs58::decode(key1).into_vec().unwrap();
        assert_eq!(secret.len(), 32);
    }
}
